package com.example.day04.controller.Operators;

import com.example.day04.controller.bean.WaterSemsor;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.Types;

public class KeyByDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2); // 并行度2， 两个子任务
        // 1> WaterSemsor(id=s2, vc=1, location=zhao)
        // 2> WaterSemsor(id=s1, vc=1, location=shengjie)  id相同的在同一个分区中，同一个任务中
        // 2> WaterSemsor(id=s1, vc=2, location=jinying)

        DataStreamSource<WaterSemsor> stream = env.fromElements(new WaterSemsor("s1", 1L, "shengjie"), new WaterSemsor("s1", 2L, "jinying"), new WaterSemsor("s2", 3L, "zhao"),  new WaterSemsor("s3", 4L, "yyyyy"));
        KeyedStream<WaterSemsor, String> waterSemsorStringKeyedStream = stream.keyBy(WaterSemsor::getId);
        // 位置只支持tupe 类型，不能是自定义pojo类型
//        SingleOutputStreamOperator<WaterSemsor> sum = waterSemsorStringKeyedStream.sum(2);
        // 使用字段名 pojo
//        SingleOutputStreamOperator<WaterSemsor> sum = waterSemsorStringKeyedStream.sum("vc");
        // max: 移动计算字段，其他字段第一次保留
//        SingleOutputStreamOperator<WaterSemsor> max = waterSemsorStringKeyedStream.max("vc");
        // maxBy: 移动计算字段，其他字段也变动
//        SingleOutputStreamOperator<WaterSemsor> max = waterSemsorStringKeyedStream.maxBy("vc");

        SingleOutputStreamOperator<WaterSemsor> reduce = waterSemsorStringKeyedStream.reduce(new ReduceFunction<WaterSemsor>() {

            @Override
            public WaterSemsor reduce(WaterSemsor value1, WaterSemsor value2) throws Exception {
                System.out.println("value1: " + value1);
                System.out.println("value2: " + value2);
                return new WaterSemsor(value1.getId(), value1.getVc() + value2.getVc(), value1.getLocation());
            }
        });
        reduce.print();
//        max.print();
//        waterSemsorStringKeyedStream.print();

        env.execute();

    }
}
